package flink.datastream;

import flink.model.Order;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.junit.Test;

import java.util.Arrays;

/**
 * @author binarylei
 * @version 2019-11-28
 */
public class SinkRedisDataStreamTest {

    @Test
    public void testUnion() throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Order> ds1 = env.fromCollection(Arrays.asList(
                new Order(1L, "order1"),
                new Order(2L, "order2")));

        FlinkJedisPoolConfig jedisPoolConfig = new FlinkJedisPoolConfig.Builder()
                .setHost("127.0.0.1").setPort(6379).build();
        RedisSink<Order> redisSink = new RedisSink<>(jedisPoolConfig, new MyRedisMapper());
        ds1.addSink(redisSink);

        env.execute();
    }

    public class MyRedisMapper implements RedisMapper<Order> {
        // redis operator
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.LPUSH);
        }

        // redis key
        @Override
        public String getKeyFromData(Order o) {
            return o.name;
        }

        // redis value
        @Override
        public String getValueFromData(Order o) {
            return o.toString();
        }
    }

}
